package com.jw.qme.socket;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.jw.qme.domain.po.Friend;
import com.jw.qme.domain.po.GroupMember;
import com.jw.qme.domain.po.Message;
import com.jw.qme.domain.po.MessageGroup;
import com.jw.qme.service.FriendService;
import com.jw.qme.service.GroupMemberService;
import com.jw.qme.service.MessageGroupService;
import com.jw.qme.service.MessageService;
import com.jw.qme.socket.message.*;
import com.jw.qme.utils.MessageProtocol;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * ClassName: com.jw.qme.socket
 * Datetime: 2025/4/4 下午9:00
 * Description:
 * 消息转发服务器
 *
 * @author 汪佳泉
 * @version 1.0
 * @since 1.0
 */
@Component
@RequiredArgsConstructor
public class SocketService implements CommandLineRunner {

    // 服务器私钥
    @Value("${socket.key.privateKey}")
    private String privateKey;

    // 服务端口
    private static final int PORT = 5209;

    // 缓冲大小
    private static final int BUFFER_SIZE = 4096 * 4;

    // Spring上下文容器
    private final ApplicationContext applicationContext;

    // 用来保存客户端的输出流和账号信息的集合
    private Map<String, SocketChannel> socketChannels;

    // 用来保存客户端和服务器间数据的对称密钥
    private Map<String, String> symmetricKeys;

    // 线程池 用来处理读处理 防止阻塞主线程 CPU密集型
    private static ThreadPoolExecutor readExecutor;

    // 线程池 用来处理用户上线 来取离线消息 IO密集型
    private static ThreadPoolExecutor offlineExecutor;

    private final MessageService messageService;

    private final MessageGroupService messageGroupService;

    private final GroupMemberService groupMemberService;

    private final FriendService friendService;

    @Override
    public void run(String... args) throws Exception {
        // 初始话线程池
        readExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2,
                Runtime.getRuntime().availableProcessors() << 1,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>());
        offlineExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() + 1,
                Runtime.getRuntime().availableProcessors() << 1,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>());
        socketChannels = (Map<String, SocketChannel>) applicationContext.getBean("socketChannels");
        symmetricKeys = (Map<String, String>) applicationContext.getBean("symmetricKeys");
        // 测试 todo 待删除
        symmetricKeys.put("user123","SyYDOwcFUURDPliG7MAkng==");
        symmetricKeys.put("user456","n8y4krega7lHSZUbitu2mg==");
        new Thread(() -> {
            try {
                startServer();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }).start();
    }

    public void startServer() throws IOException {
        // 1.创建 Selector
        Selector selector = Selector.open();
        // 2.创建 ServerSocketChannel 并配置未非阻塞
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(PORT)); // 绑订端口
        serverSocketChannel.configureBlocking(false); // 设置未非阻塞
        // 3.将 ServerSocketChannel 注册到 Selector 中 监听 OP_ACCEPT事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服务器启动成功，监听端口：" + PORT);
        // 4.循环处理事件
        while (true) {
            // 5.轮询等待事件 会阻塞直到有事件
            selector.select();
            // 6.获取所有准好的待处理事件
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                try {
                    if (key.isAcceptable()) {
                        // 新创建socket连接
                        handleAccept(key, selector);
                    } else if (key.isReadable()) {
                        // 处理读事件
                        readExecutor.submit(() -> {
                            try {
                                handleRead(key);
                            } catch (IOException e) {
                                // 关闭连接通道 并取消注册
                                key.cancel();
                                if (key.channel() != null) {
                                    try {
                                        key.channel().close();
                                    } catch (IOException ex) {
                                        throw new RuntimeException(ex);
                                    }
                                }
                            }
                        });
                         // handleRead(key); // 非异步 便于开发调试
                    } else if (key.isWritable()) {
                        // 处理写事件
                    }
                } catch (Exception e) {
                    // 关闭连接通道 并取消注册
                    key.cancel();
                    if (key.channel() != null) {
                        key.channel().close();
                    }
                    e.printStackTrace();
                } finally {
                    // 移除已处理的事件
                    iterator.remove();
                }
            }
        }
    }

    /**
     * 创建新的连接请求
     * @param key
     * @param selector
     * @throws IOException
     */
    private void handleAccept(SelectionKey key, Selector selector) throws IOException {
        ServerSocketChannel serverSocket = (ServerSocketChannel) key.channel();
        SocketChannel socketChannel = serverSocket.accept();
        if (socketChannel != null) {
            // 注册到对应的选择器上
            // 设置为非阻塞
            socketChannel.configureBlocking(false);
            // 注册到selector上 并监听 OP_READ 读事件
            socketChannel.register(selector, SelectionKey.OP_READ);
        }
    }

    // 处理读事件
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
        int read = client.read(buffer);
        if (read == -1) {
            key.cancel();
            client.close();
            System.out.println("客户端已断开连接：" + client.getRemoteAddress());
            return;
        }
        if (read > 0) {
            buffer.flip();
            // 使用协议解析消息
            MessageProtocol.DecodedMessage decodedMessage = MessageProtocol.decode(buffer);
            if (decodedMessage == null) {
                System.out.println("收到不完整数据，等待下一次读取");
                return;
            }
            byte[] bytes = decodedMessage.getContent();
            String jsonString = new String(bytes, StandardCharsets.UTF_8);
            SocketMessage socketMessage = null;
            if(decodedMessage.getType() != 0x05){
                socketMessage = JSON.parseObject(jsonString, SocketMessage.class);
            }

            byte type = decodedMessage.getType();
            MessageType msgType = MessageType.fromValue(type);
            if (msgType == null) {
                System.err.println("未知的消息类型：" + type);
                return;
            }
            switch (msgType) {
                case USER_LOGIN:
                    bandChannelWithUser(jsonString,client);
                    break;
                case MESSAGE_TO:
                    ChatMessage message = (ChatMessage) MessageFactory.getMessage(socketMessage);
                    sendMessage(message,MessageType.MESSAGE_TO);
                    break;
                case MESSAGE_GROUP:
                    GroupMessage groupMessage = (GroupMessage) MessageFactory.getMessage(socketMessage);
                    sendMessageGroup(groupMessage);
                    break;
                case EMOJI_IMAGE:
                    ChatMessage emoMessage = (ChatMessage) MessageFactory.getMessage(socketMessage);
                    sendMessage(emoMessage,MessageType.EMOJI_IMAGE);
                    break;
                default:
                    System.err.println("未知的消息类型：" + msgType);
            }

        }
    }

    /**
     * 将用户与连接进行绑定
     * @param jsonString
     * @param client
     */
    private void bandChannelWithUser(String jsonString,SocketChannel client){
        try {
            socketChannels.put(jsonString, client);
            // 进行数据库中添加好友申请的拉取 时间较长用IO线程池处理
            offlineExecutor.submit(()->{
                // 去数据库中进行查询 待处理的好友请求 然后发送一个通知
                List<Friend> friendByUsername = friendService.getFriendByUsername(jsonString, true);
                if(friendByUsername!= null && friendByUsername.size() != 0){
                    sendMessage(jsonString,jsonString,"您有" + friendByUsername.size() + "条好友请求等待您的处理",MessageType.NOTIFICATION);
                }
            });
            System.out.println("客户端已经成功连接：" + client.getRemoteAddress());
        } catch (Exception e) {
            System.err.println("绑定用户名失败：" + e.getMessage());
        }
    }

    /**
     * 统一消息发送
     * @param message
     * @param type
     */
    private void sendMessage(ChatMessage message,MessageType type) {
        String receiverId = message.getReceiverId();
        SocketChannel client = socketChannels.get(receiverId);
        // 判断当前消息是否为群聊消息
        boolean isGroup = message.getGroupId() == null ? false : true;
        if (client != null && client.isOpen()) {
            try {
                String jsonContent;
                // 构建消息内容
                ByteBuffer encodedBuffer;

                switch(type){
                    case MESSAGE_TO:
                        // 单聊消息
                        jsonContent = JSON.toJSONString(message);
                        encodedBuffer = MessageProtocol.encode(MessageType.MESSAGE_TO.getValue(), jsonContent.getBytes(StandardCharsets.UTF_8));
                        break;
                    case MESSAGE_GROUP:
                        // 群聊消息
                        GroupMessage groupMessage = new GroupMessage();
                        BeanUtils.copyProperties(message, groupMessage);
                        jsonContent = JSON.toJSONString(groupMessage);
                        encodedBuffer = MessageProtocol.encode(MessageType.MESSAGE_GROUP.getValue(), jsonContent.getBytes(StandardCharsets.UTF_8));
                        break;
                    case FRIEND_REQUEST:
                        jsonContent = JSON.toJSONString(message);
                        encodedBuffer = MessageProtocol.encode(MessageType.FRIEND_REQUEST.getValue(), jsonContent.getBytes(StandardCharsets.UTF_8));
                        break;
                    case EMOJI_IMAGE:
                        jsonContent = JSON.toJSONString(message);
                        encodedBuffer = MessageProtocol.encode(MessageType.EMOJI_IMAGE.getValue(), jsonContent.getBytes(StandardCharsets.UTF_8));
                        break;
                    case NOTIFICATION:
                        jsonContent = JSON.toJSONString(message);
                        encodedBuffer = MessageProtocol.encode(MessageType.NOTIFICATION.getValue(), jsonContent.getBytes(StandardCharsets.UTF_8));
                        break;
                    default:
                        throw new RuntimeException("未知的消息类型：" + type);
                }

                // 发送消息
                while (encodedBuffer.hasRemaining()) {
                    client.write(encodedBuffer);
                }

                System.out.println("消息发送成功：" + jsonContent);

                if(!isGroup) writeChatMessage2DB(message, false);

            } catch (IOException e) {
                socketChannels.remove(receiverId);
                throw new RuntimeException(e);
            }
        } else {
            if(!isGroup) writeChatMessage2DB(message, false);
        }
    }

    /**
     * 发送群聊消息
     * @param message
     */
    private void sendMessageGroup(GroupMessage message) {
        // 去数据库中查找对应的用户ID
        List<GroupMember> list = groupMemberService.list(new LambdaQueryWrapper<GroupMember>().eq(GroupMember::getGroupId, message.getGroupId()).ne(GroupMember::getUsername, message.getSenderId()));
        // 遍历这个列表
        for (GroupMember member : list) {
            // 封装为单条消息
            ChatMessage chatMessage = new ChatMessage();
            BeanUtils.copyProperties(message, chatMessage);
            chatMessage.setReceiverId(member.getUsername());
            sendMessage(chatMessage,MessageType.MESSAGE_GROUP);
        }
        // 将消息持久话到数据库中
        writeGroupMessage2DB(message,false);
    }

    /**
     * 将消息保存到数据库中 当用户启动时 将消息推动到用户的客户端中
     * @param message 消息
     * @param flag    标记是否为离线消息
     */
    private void writeChatMessage2DB(ChatMessage message, Boolean flag) {
        messageService.save(Message.builder()
                .senderId(message.getSenderId())
                .receiverId(message.getReceiverId())
                .content(JSON.toJSONString(message))
                .isRead(flag ? false : true)
                .build());
    }

    /**
     * 将消息保存到数据库中 当用户启动时 将消息推动到用户的客户端中
     * @param message 消息
     * @param flag    标记是否为离线消息
     */
    private void writeGroupMessage2DB(GroupMessage message, Boolean flag) {
        messageGroupService.save(MessageGroup.builder()
                .senderId(message.getSenderId())
                .groupId(message.getGroupId())
                .content(JSON.toJSONString(message))
                .isRead(flag ? false : true)
                .build());
    }

    /**
     * 对外提供的统一推送消息方法
     * @param sendName
     * @param receiveName
     * @param content
     */
    public void sendMessage(String sendName ,String receiveName,String content,MessageType type) {
        // 将本消息封装为 ChatMessage 对象
        ChatMessage chatMessage = new ChatMessage();
        chatMessage.setSenderId(sendName);
        chatMessage.setReceiverId(receiveName);
        chatMessage.setContent(content);
        sendMessage(chatMessage,type);
    }

}
